package com.atguigu.app.dwd.db;

import com.atguigu.Util.MyKafkaUtil;
import com.atguigu.Util.MySqlUtil;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author hjy
 * @create 2023/3/11 15:23
 */
//数据流:app/客户端->mysql->maxwell->kafka(topic_db  ods)->Dwd02_TradeCartAdd->kafka(dwd)
//程 序:mock->mysql->maxwell->kafka(zk)->Dwd02_TradeCartAdd->kafka(zk)
public class Dwd02_TradeCartAdd {
    public static void main(String[] args) {
        //todo 1 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        env.enableCheckpointing(5000);
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall-flink/check");
//        env.getCheckpointConfig().setCheckpointTimeout(60000);
//        env.setStateBackend(new HashMapStateBackend());
        //        System.setProperty("HADOOP_USER_NAME", "atguigu");
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //todo 2 从kafka读数据 topic_db 建表
//        System.out.println(MyKafkaUtil.getKafkaDDL("tradeCartAdd"));
        tableEnv.executeSql(MyKafkaUtil.getTopicDBDDL("tradeCartAdd"));
        //打印测试
//        tableEnv.executeSql("select * from topic_db").print();
        //todo 3 过滤出加购数据
        Table tradeCart = tableEnv.sqlQuery("select\n" +
                "   `data`['id'] id,\n" +
                "   `data`['user_id'] user_id,\n" +
                "   `data`['sku_id'] sku_id,\n" +
                "   `data`['cart_price'] cart_price,\n" +
                "   `data`['sku_num'] sku_num,\n" +
                "   `data`['img_url'] img_url,\n" +
                "   `data`['sku_name'] sku_name,\n" +
                "   `data`['is_checked'] is_checked,\n" +
                "   `data`['create_time'] create_time,\n" +
                "   `data`['operate_time'] operate_time,\n" +
                "   `data`['is_ordered'] is_ordered,\n" +
                "   `data`['order_time'] order_time,\n" +
                "   `data`['source_type'] source_type,\n" +
                "   `data`['source_id'] source_id,\n" +
                "   pt\n" +
                "from\n" +
                "  topic_db\n" +
                "where\n" +
                "  `database`='gmall' and `table`='cart_info'\n" +
                "  and `type`='insert'");
        tableEnv.createTemporaryView("tradeCart",tradeCart);
//        tableEnv.executeSql("select * from tradeCart").print();
        //todo 4 建立维表
        //使用jdbc连接从mysql数据库中查询 查询base_dic表的数据
        //使用executeSql方法可以直接将查询到的数据构建成表，面对这个方法构建出来的表想要查询到里面的数据，需要使用sqlQuery方法来查询，这种方法查出来的返回的是一个table对象，所以需要建临时表
        /*总结
            使用 executeSql方法没有返回值，可以用来查询外部数据
            使用sqlQuery方法可以返回一个表，用来查询传进来的数据，并返回一个表 但这个表最好构建一个临时表来方便使用
         */
        //todo 从mysql查数据
        tableEnv.executeSql(MySqlUtil.getMysqlDDl());


        //todo 5 两表关联
        Table resultTable = tableEnv.sqlQuery("select\n" +
                "    cart.id,\n" +
                "    cart.user_id,\n" +
                "    cart.sku_id,\n" +
                "    cart.cart_price,\n" +
                "    cart.sku_num,\n" +
                "    cart.img_url,\n" +
                "    cart.sku_name,\n" +
                "    cart.is_checked,\n" +
                "    cart.create_time,\n" +
                "    cart.operate_time,\n" +
                "    cart.is_ordered,\n" +
                "    cart.order_time,\n" +
                "    cart.source_type,\n" +
                "    dic.dic_name,\n" +
                "    cart.source_id\n" +
                "from\n" +
                "    tradeCart AS cart\n" +
                "join \n" +
                "    base_dic FOR SYSTEM_TIME AS OF cart.pt AS dic\n" +
                "on\n" +
                "    cart.source_type=dic.dic_code");
        //建立临时表
        tableEnv.createTemporaryView("resultTable",resultTable);
        //todo 6 建立dwd_tradeCart_add 加购事实表
        tableEnv.executeSql("create table dwd_cart_info(\n" +
                "  `id` STRING,\n" +
                "  `user_id` STRING,\n" +
                "  `sku_id` STRING,\n" +
                "  `cart_price` STRING,\n" +
                "  `sku_num` STRING,\n" +
                "  `img_url` STRING,\n" +
                "  `sku_name` STRING,\n" +
                "  `is_checked` STRING,\n" +
                "  `create_time` STRING,\n" +
                "  `operate_time` STRING,\n" +
                "  `is_ordered` STRING,\n" +
                "  `order_time` STRING,\n" +
                "  `source_type` STRING,\n" +
                "  `dic_name` STRING,\n" +
                "  `source_id` STRING\n" +
                ")"+MyKafkaUtil.getKafkaSinkDDL("dwd_cart_info"));
        //todo 7 写入数据
        tableEnv.executeSql("insert into dwd_cart_info select * from resultTable");
    }
}
